package com.ruoyi.eqmonitor.dispose.mapper;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONArray;
import cn.hutool.json.JSONObject;
import com.baomidou.lock.LockInfo;
import com.baomidou.lock.LockTemplate;
import com.baomidou.lock.executor.RedisTemplateLockExecutor;
import com.ruoyi.common.core.utils.SpringUtils;
import com.ruoyi.common.core.utils.StringUtils;
import com.ruoyi.common.redis.service.RedisService;
import com.ruoyi.eqmonitor.dispose.domain.ExteriorMonitorData;
import com.ruoyi.eqmonitor.dispose.domain.OpentsdbTags;
import com.ruoyi.eqmonitor.dispose.service.IExteriorMonitorDataService;
import com.ruoyi.eqmonitor.dispose.service.IotService;
import com.ruoyi.eqmonitor.domain.OptDeviceMonitorLastData;
import com.ruoyi.eqmonitor.dtos.OptDeviceDto;
import com.ruoyi.eqmonitor.exception.OpentsdbException;
import com.ruoyi.eqmonitor.exception.OpentsdbValidateException;
import com.ruoyi.eqmonitor.service.IOptDeviceService;
import com.ruoyi.eqmonitor.utils.Constant;
import com.ruoyi.eqmonitor.utils.OpenTSDBUtils;
import lombok.extern.slf4j.Slf4j;
import org.opentsdb.client.bean.request.Point;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * 监测数据写入与重试
 *
 * @author dh
 */
@Slf4j
@Component
public class MonitorDataMapper {

    /**
     * 最新实时数据Redis key
     */
    private static final String LAST_MONITOR_DATA_REDIS_KEY = "last_monitor_data:";
    /**
     * 防止重复数据Redis key
     */
    public static final String REPETITION_MONITOR_KEY = "repetition_monitor_data:";
    @Resource
    private OpenTSDBUtils openTSDBUtils;

    @Value("${spring.iotdb.enable-save:false}")
    private Boolean enableSave;

    @Resource
    private IOptDeviceService iOptDeviceService;
    @Resource
    private RedisService redisService;
    @Resource
    private IExteriorMonitorDataService iExteriorMonitorDataService;
    @Resource
    private LockTemplate lockTemplate;

    private static final String LOCK_TEMPLATE_KEY = "{}_opentsdb_save_lock";

    /**
     * 写入监测数据
     * 重试3次, 第一次重试为2秒，第二次为2秒，第三次为2秒 不包含 OpentsdbValidateException
     * 存储格式:
     * [
     *  {
     *    "metric": "monitor.equipment_tzsw",
     *    "timestamp": 1346846400, // 秒级时间戳
     *    "value": 13.72, // 监测指标对应的数值
     *    "tags": { //
     *       "snCode": "0C21JD",
     *       "monitorName": "设备实时电压",
     *       dataUnit: "V",
     *       equipmentType: "流量电导一体机",
     *       monitorCode: "0x0501"
     *    }
     *  }
     * ]
     */
    public void write(String snCode, OptDeviceDto optDevice, JSONArray messageArray, String version) throws OpentsdbException {
        log.info("enableSave:{}",enableSave);
        // 不允许存储 则放弃
        if (!enableSave) return;

        String insertDeviceCode = optDevice.getDeviceCode();
        // 1. 校验消息体是否为空
        if (messageArray == null || messageArray.isEmpty()) {
            throw new OpentsdbValidateException(StrUtil.format("snCode:{}, deviceCode: {}, 消息流水: {}, 写入db失败, 消息体不存在或为空: {}", snCode, insertDeviceCode, version, messageArray));
        }
        if (StrUtil.isBlank(insertDeviceCode)) {
            throw new OpentsdbValidateException(StrUtil.format("snCode:{}, deviceCode: {}, 消息流水: {}, 写入db失败, deviceCode为空: {}", snCode, insertDeviceCode, version, messageArray));
        }

        // 2. 批量处理数据
        String lockKey = StrUtil.format(LOCK_TEMPLATE_KEY, insertDeviceCode);
        // 获取锁
        final LockInfo lockInfo = lockTemplate.lock(lockKey, 120000L, 150000L, RedisTemplateLockExecutor.class);
        if (null == lockInfo) {
            throw new RuntimeException(StrUtil.format("snCode:{}, deviceCode: {}, 消息流水: {}, 执行存储获取锁失败", snCode, insertDeviceCode, version));
        }
        // 3. 处理数据格式
        try {
            // 去重使用map
            Map<String, OptDeviceMonitorLastData> duplication = new HashMap<>();
            // mysql数据使用
            List<ExteriorMonitorData> monitorDataList = new ArrayList<>();
            // opentsdb数据使用
            List<Point> pointList = new ArrayList<>();
            // redis数据使用
            List<String> repetitionRedisKeys = new ArrayList<>();
            for (Object o : messageArray) {
                try {
                    JSONObject message = (JSONObject) o;
                    Constant.MonitorInfo info = Constant.MonitorInfo.getInfoByCode(message.getStr("monitorCode"));
                    // 3.1 没有这个监测指标 或者这个监测指标不允许存储 则抛弃
                    if (ObjectUtil.isNull(info) || !info.isSave()) {
                        continue;
                    }
                    // 3.2 日期不存在 则抛弃
                    String monitorTime = message.getStr("monitorTime", null);
                    if (StrUtil.isEmpty(monitorTime)) {
                        continue;
                    }

                    // 3.3 redis中存在这个数据 则抛弃
                    DateTime parseDate = DateUtil.parse(monitorTime);
                    String repetitionRedisKey = getRepetitionRedisKey(insertDeviceCode, info.getCode(), parseDate);
                    if (redisService.hasKey(repetitionRedisKey)) {
                        continue;
                    }

                    Double monitorData = message.getDouble("monitorData", 0D);
                    // 3.4 校验并创建tags对象 如果转换失败 则抛弃
                    OpentsdbTags opentsdbTags = validAndBuildTags(optDevice, message);
                    if (opentsdbTags == null) continue;

                    // 3.5 创建mysql需要的exteriorMonitorData对象
                    ExteriorMonitorData exteriorMonitorData = ExteriorMonitorData.buildByOpentsdbTags(parseDate, monitorData, opentsdbTags);
                    // 3.6 创建opentsdb需要的point对象 [metric, 时间戳, tag值]
                    Point point = openTSDBUtils.buildPoint(parseDate, monitorData, opentsdbTags);
                    // 3.7 计算每个指标最后一个监测数据 注入设备最后监测时间
                    OptDeviceMonitorLastData lastData = duplication.get(opentsdbTags.getMonitorCode());
                    OptDeviceMonitorLastData currentData = buildDeviceMonitorLastData(optDevice, opentsdbTags, parseDate, monitorData);
                    if (lastData == null) {
                        duplication.put(opentsdbTags.getMonitorCode(), currentData);
                    } else if (DateUtil.compare(lastData.getMonitorTime(), currentData.getMonitorTime()) < 0) {
                        duplication.put(opentsdbTags.getMonitorCode(), currentData);
                    }

                    // 3.8 注入mysql需要数据结构
                    monitorDataList.add(exteriorMonitorData);
                    // 3.9 注入opentsdb需要数据结构
                    pointList.add(point);
                    // 3.10 注入redis需要数据结构
                    repetitionRedisKeys.add(repetitionRedisKey);
                } catch (Exception e) {
                    log.error("snCode:{}, deviceCode: {}, 消息流水: {}, 解析监测数据失败", snCode, insertDeviceCode, version, e);
                }
            }
            // 4. 执行存储
            SpringUtils.getBean(this.getClass()).saveMonitorData(insertDeviceCode, version, duplication, monitorDataList, pointList, repetitionRedisKeys);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            //释放锁
            lockTemplate.releaseLock(lockInfo);
        }
    }
    @Resource
    private IotService iotService;
    /**
     * 存储需要的内容
     * @param insertDeviceCode 设备deviceCode
     * @param version 消息流水
     * @param duplication 去重后的最新数据
     * @param monitorDataList 监测数据列表 (存储mysql用)
     * @param pointList 监测数据列表 (存储opentsdb用)
     * @param repetitionRedisKeys 监测数据keys (redis去重用)
     * @throws OpentsdbException opentsdb存储失败
     */
    @Transactional(rollbackFor = Exception.class)
    public void saveMonitorData(String insertDeviceCode,
                                String version,
                                Map<String, OptDeviceMonitorLastData> duplication,
                                List<ExteriorMonitorData> monitorDataList,
                                List<Point> pointList,
                                List<String> repetitionRedisKeys) throws OpentsdbException {
        if (CollectionUtil.isNotEmpty(pointList)) {
            // 1. 序列化最新数据
            saveOrUpdateMonitorLastData(insertDeviceCode, duplication.values());
            // 2. 入库mysql
            iExteriorMonitorDataService.insertBeach(monitorDataList);
            // 3. 执行iotDB存储
            iotService.write(pointList, version, insertDeviceCode);
            // 4. 都存储成功 则向入redis中注入此键, 防止重复, 保存25小时 (还有28日14:03接到27日23:13数据的情况,所以需要加时间....)
            for (String repetitionRedisKey : repetitionRedisKeys) {
                redisService.setCacheObject(repetitionRedisKey, StringUtils.EMPTY, 25L, TimeUnit.HOURS);
            }
        }
    }

    /**
     * 更新最后的监测数据
     * @param snCode 设备编码
     * @param listData 最后监测数据列表
     */
    private void saveOrUpdateMonitorLastData(String snCode, Collection<OptDeviceMonitorLastData> listData) {
        String redisHKey = getRedisHKey(snCode);
        // 原始数据
        Map<String, OptDeviceMonitorLastData> cacheMap = redisService.getCacheMap(redisHKey);
        // 分组后的新数据
        Map<String, OptDeviceMonitorLastData> groupBy = listData.stream()
                .collect(Collectors.toMap(OptDeviceMonitorLastData::getCode, Function.identity()));
        if (cacheMap == null) {
            redisService.setCacheMap(redisHKey, groupBy);
        } else {
            for (Map.Entry<String, OptDeviceMonitorLastData> entry : groupBy.entrySet()) {
                OptDeviceMonitorLastData lastData = cacheMap.get(entry.getKey());
                // hash中不存在此监测指标 或者存在但时间小于最新的指标
                if (lastData == null || DateUtil.compare(lastData.getMonitorTime(), entry.getValue().getMonitorTime()) < 0) {
                    redisService.setCacheMapValue(redisHKey, entry.getKey(), entry.getValue());
                }
            }
        }
    }

    /**
     * 最后监测数据的hKey
     * @param snCode 设备编码
     * @return redis hKey
     */
    public static String getRedisHKey(String snCode) {
        return LAST_MONITOR_DATA_REDIS_KEY.concat(snCode);
    }

    /**
     * 创建数据库需要的 OptDeviceMonitorLastData 对象
     * @param optDevice 设备信息
     * @param opentsdbTags opentsdb数据项
     * @param monitorTime 数据时间
     * @param monitorData 数据值
     * @return OptDeviceMonitorLastData对象
     */
    private OptDeviceMonitorLastData buildDeviceMonitorLastData(OptDeviceDto optDevice, OpentsdbTags opentsdbTags, Date monitorTime, Double monitorData) {
        OptDeviceMonitorLastData deviceMonitorLastData = new OptDeviceMonitorLastData();
        deviceMonitorLastData.setDeviceName(optDevice.getDeviceName());
        deviceMonitorLastData.setDeviceCode(optDevice.getDeviceCode());
        deviceMonitorLastData.setSiteName(optDevice.getSiteName());
        deviceMonitorLastData.setDeviceType(optDevice.getDeviceType());
        deviceMonitorLastData.setInstallTime(optDevice.getInstallTime());
        deviceMonitorLastData.setAbx(optDevice.getAbx());
        deviceMonitorLastData.setAby(optDevice.getAby());
        deviceMonitorLastData.setCode(opentsdbTags.getMonitorCode());
        deviceMonitorLastData.setCodeValue(monitorData);
        deviceMonitorLastData.setDataUnit(opentsdbTags.getDataUnit());
        deviceMonitorLastData.setMonitorTime(monitorTime);
        deviceMonitorLastData.setMonitorName(opentsdbTags.getMonitorName());
        return deviceMonitorLastData;
    }


    /**
     * 校验并构建tags对象
     * @param optDevice 设备对象
     * @param message 设备消息
     * @return tags对象
     */
    private OpentsdbTags validAndBuildTags(OptDeviceDto optDevice, JSONObject message) {
        // 1. 取出并校验数据
        String snCode = optDevice.getDeviceCode();
        String monitorCode = message.getStr("monitorCode");
        Constant.MonitorInfo monitorInfo;
        Constant.MonitorInfo monitorCodeEnum = Constant.MonitorInfo.getMonitorInfoByCode(monitorCode);
        Constant.MonitorInfo paramCodeEnum = Constant.MonitorInfo.getParamInfoByCode(monitorCode);
        monitorInfo = monitorCodeEnum == null ? paramCodeEnum : monitorCodeEnum;
        if (monitorInfo == null || !monitorInfo.isSave()) {
            if (monitorInfo == null) {
                // 这里不存在的监测指标, 打一个日志后放弃处理
                log.warn("监测指标不存在: {}, 数据: {}", monitorCode, message);
            }
            return null;
        }
        // 注意: 不要轻易使用设备回传的如下两个参数 可能与实际参数定义产生差别, 需要自行维护监测项名称
        // 监测指标名称
        String dataUnit = message.getStr("dataUnit", monitorInfo.getUnit());
        // 监测指标单位
        String monitorName = StrUtil.isEmpty(monitorInfo.getName()) ?
                message.getStr("monitorName", StrUtil.EMPTY) : monitorInfo.getName();


        // 监测数据值
        Double monitorData = message.getDouble("monitorData");

        if (StrUtil.isEmpty(monitorCode)) {
            throw new OpentsdbValidateException(snCode + " 解析监测数据失败, 监测指标不存在" + message);
        } else if (ObjectUtil.isNull(monitorData)) {
            throw new OpentsdbValidateException(snCode + " 解析监测数据失败, 监测数据值不存在" + message);
        }

        // 2. 将数据序列化为opentsdb需要的格式
        OpentsdbTags opentsdbTags = new OpentsdbTags();
        opentsdbTags.setSnCode(snCode);
        opentsdbTags.setMonitorCode(monitorCode);
        opentsdbTags.setMonitorName(monitorName);
        opentsdbTags.setDataUnit(dataUnit);
        String equipmentType = Constant.EquipmentType.getValueByCode(optDevice.getDeviceType());
        opentsdbTags.setEquipmentType(equipmentType);
        opentsdbTags.setEquipmentTypeCode(optDevice.getDeviceType());

        return opentsdbTags;
    }

    /**
     * 获取去重redis 的键
     */
    public static String getRepetitionRedisKey(String snCode, String monitorCode, DateTime dateTime) {
        return REPETITION_MONITOR_KEY.concat(snCode).concat(":").concat(monitorCode).concat(":").concat(dateTime.toDateStr()).concat(":").concat(dateTime.toTimeStr().replace(":", "_").concat(":"));
    }
}
